package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.constant.MqttDestinationConst;
import cn.tedu.charging.common.pojo.entity.ChargingProgressEsPO;
import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.common.pojo.message.ProgressData;
import cn.tedu.charging.common.pojo.param.ProgressCostParam;
import cn.tedu.charging.common.pojo.vo.ProgressCostVO;
import cn.tedu.charging.common.protocol.JsonResult;
import cn.tedu.charging.common.protocol.WebSocketMessage;
import com.alibaba.fastjson2.JSON;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.BeanUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

@Slf4j
@AllArgsConstructor
@NoArgsConstructor
public class MyMqttCallback implements MqttCallbackExtended {

    private MqttClient mqttClient;

    private MqttContext mqttContext;

    /**
     *调用时机: 当前回调对象所属的连接mqttClient成功连接完毕的时候
     *reconnec: 表示本次连接成功是重连,还是第一次连 false第一次 true重连
     *serverURI: 连接目标地址 address相同
     *当放大调用 代表连接建立成功,可以实现初始化,订阅的操作
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if (reconnect){
            log.info("重新连接,emqx地址:{}",serverURI);
        }else{
            log.info("第一次连接,emqx地址:{}",serverURI);
        }
        //当前回调对象 所属的mqttClient监听订阅主题
        try{
            //普通订阅 群发
            //mqttClient.subscribe("topic01");
            //mqttClient.subscribe("topic02");
            //共享订阅 争抢
            /*mqttClient.subscribe("$share/test/topic01");
            mqttClient.subscribe("$share/test/topic02");*/
            //共享订阅 第一个主题 就是设备自检反馈结果
            mqttClient.subscribe("$share/order/"+ MqttDestinationConst.GUN_CHECK_RESULT_TOPIC);
            //共享订阅 订阅第二个主题 充电进度
            mqttClient.subscribe("$share/order/"+ MqttDestinationConst.CHARGING_PROGRESS_TOPIC);
        }catch (Exception e){
            log.error("订阅主题失败",e);
        }
    }

    /**
     *时间节点: 当前回调对象所属的mqttClient断开连接的时候调用
     *cause: 断开连接的原因,超时,拒绝
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.info("断开连接,原因:{}",cause.getMessage());
    }

    /**
     *调用时机: 当前所属的连接 投递发送消息成功会调用
     *token: 投递消息的完整镜像,可以通过token获取投递消息的所有内容
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("消息投递成功");
    }

    /**
     *回调时机: 当前所属连接 监听订阅的主题出现新的消息时,会获取消息,调用messageArrived
     *topic: 消息主题来源名字
     *message:消息对象包含消息头消息体消息属性
     *消费功能 可以在messageArrived中完成
     *
     * 设备 自检消息 到订单
     * 设备 充电进度消息 到订单
     * 是通过Topic 区分
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("从:{}主题,拿到消息:{}",topic,new String(message.getPayload()));
        //1.回调所属的连接 同时订阅了多个主题 判断
        if (topic != null && topic.equals(MqttDestinationConst.GUN_CHECK_RESULT_TOPIC)){
            //设备反馈自检结果
            //2.消息是否是空
            //byte[] 字节数组
            byte[] payload = message.getPayload();
            if (payload!=null&&payload.length>0){
                //3有消息 二进制反序列化成 String 对象 new String(byte[])
                String json = new String(payload, StandardCharsets.UTF_8);
                //把字符串转化为对象  ChargingCheckResultMessage
                ChargingCheckResultMessage msg = JSON.parseObject(json, ChargingCheckResultMessage.class);
                //调用业务处理 mqttConsumerService
                //1.添加抢锁逻辑
                String lockKey="charging:order:lock:"+msg.getOrderNo();
                RedisTemplate redisTemplate = mqttContext.getRedisTemplate();
                ValueOperations stringOps = redisTemplate.opsForValue();
                boolean haveLock=false;
                int count=0;
                try{
                    do{
                        count++;//第一次抢 count==1 2 3 4 第四次最高了
                        if (count>1&&count<=4){
                            //不是第一次抢
                            Thread.sleep(1000);
                        }else if(count>4){
                            //5,6,7不让抢 消费结束了
                            return;
                        }
                        haveLock=stringOps.setIfAbsent(lockKey,"",5, TimeUnit.SECONDS);
                    }while (!haveLock);
                    //调用业务代码 处理业务逻辑
                    mqttContext.getConsumerService().handlerCheckResult(msg);
                    //业务处理完毕之后,手动确认反馈
                }catch (Exception e){
                    log.error("抢锁执行业务异常",e);
                }finally {
                    redisTemplate.delete(lockKey);
                }

            }
        }else if (topic != null && topic.equals(MqttDestinationConst.CHARGING_PROGRESS_TOPIC)){
            //处理 设备同步的充电进度数据
            //message 获取 字节数组
            byte[] payload = message.getPayload();
            if (payload!=null&&payload.length>0){
                //把字节数组转化为字符串 new String
                String str = new String(payload, StandardCharsets.UTF_8);
                //把字符串转化为 ChargingProgressMsg 对象 JSON.parseObject(str,对象.class)
                ChargingProgressMsg chargingProgressMsg =
                        JSON.parseObject(str, ChargingProgressMsg.class);
                handleProcessMsg(chargingProgressMsg);
            }
        }
    }

    /**
     * 处理充电进度消息
     * 自顶向下编程
     */
    private void handleProcessMsg(ChargingProgressMsg chargingProgressMsg){
        log.debug("处理充电进度消息:{}",chargingProgressMsg);
        //1 远程调用计价服务 计算价格
        ProgressCostVO progressCostVO = getOrderCost(chargingProgressMsg);
        //2 保存 充电进度 和 价格信息 到 ES
        saveChargingProgressToES(chargingProgressMsg,progressCostVO);
        //3 根据充电进度判断是否完成
        if (chargingProgressMsg.getIsFull()){
            //3.1 充满 推送充满的信息给用户
            WebSocketMessage<String> chargingFullMsg = new WebSocketMessage<>();
            chargingFullMsg.setState(1);
            chargingFullMsg.setData("A已充满,订单完成");
            //推送给用户
            mqttContext.getWebSocketServer().sendMsg2User(
                    chargingProgressMsg.getUserId(),chargingFullMsg);
        }else {
            //3.2 没充满 推送充电进度给用户
            WebSocketMessage webSocketProcessMessage = new WebSocketMessage<>();
            //充电进度
            webSocketProcessMessage.setState(3);
            webSocketProcessMessage.setMessage("充电中");

            //创建 同步给用户的充电进度数据对象 ProgressData
            ProgressData progressData = new ProgressData();
            //1 充电进度 总充满的需要充的度数 当前充电度数
            //思考 总充满的度数 从哪来 充电枪插入到车 读取车的数据 电池容量 电池剩余电量
            progressData.setTotalCapacity(100.00);
            //设置当前充电度数
            progressData.setChargingCapacity(chargingProgressMsg.getChargingCapacity());
            //2 总花费 从计价服务中获取
            progressData.setTotalCost(progressCostVO.getTotalCost());
            //3 充电时长 当前时间 减去 第一次同步的时间 利于java的时间API计算
            //小时
            progressData.setHours(0);
            //分钟
            progressData.setMinutes(5);
            //秒
            progressData.setSeconds(15);

            //把 充电数据 设置到 websocket消息体
            webSocketProcessMessage.setData(progressData);

            //推送给用户
            mqttContext.getWebSocketServer().sendMsg2User(
                    chargingProgressMsg.getUserId(),webSocketProcessMessage);
        }
    }

    /**
     * 保存 充电进度 和 价格信息 到 ES
     * @param chargingProgressMsg 充电进度
     * @param progressCostVO  价格信息
     */
    private void saveChargingProgressToES(ChargingProgressMsg chargingProgressMsg,
                                          ProgressCostVO progressCostVO){
        //创建ES要保存的充电数据 PO
        ChargingProgressEsPO chargingProgressEsPO = new ChargingProgressEsPO();
        //充电进度 复制到 chargingProgressEsPO
        BeanUtils.copyProperties(chargingProgressMsg,chargingProgressEsPO);
        //价格信息 复制到 chargingProgressEsPO
        BeanUtils.copyProperties(progressCostVO,chargingProgressEsPO);
        //id设置为 当前时间戳
        //时序数据 时序数据库 InfluxDB。
        //数据特点 高频插入 低频更新和删除 查询
        // 充电数据
        // 时间 顺序 数据
        // 15:31 4
        // 15:32 5
        //天气预报
        // 15:31 31度
        // 15:32 31.5度
        chargingProgressEsPO.setId(System.currentTimeMillis()+"");
        //保存
        mqttContext.getProcessEsRepository().save(chargingProgressEsPO);
    }

    /**
     * 获取订单价格
     */
    private ProgressCostVO getOrderCost(ChargingProgressMsg chargingProgressMsg){
        ProgressCostParam costParam = new ProgressCostParam();
        //chargingProgressMsg 属性 复制到 costParam
        BeanUtils.copyProperties(chargingProgressMsg,costParam);
        JsonResult<ProgressCostVO> result = mqttContext.getCostClient().calculateCost(costParam);
        if (result != null) {
            return result.getData() ;
        }
        return null;
    }
}
